package com.zuikaku.sink;

import com.zuikaku.pojo.Order;
import com.zuikaku.util.JDBCUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.PreparedStatement;

public class OrderSink extends RichSinkFunction<Order> {

    /**
     * sink启动时的操作
     * @param parameters The configuration containing the parameters attached to the contract.
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化JDBC
        JDBCUtil.init();
    }

    /**
     * sink关闭时的操作
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        JDBCUtil.closeAllConnection();
    }

    /**
     * 对每个数据的处理
     * @param order
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(Order order, Context context) throws Exception {
        String sql = "INSERT INTO t_order VALUES (?,?,?,?)";
        PreparedStatement preparedStatement = JDBCUtil.getConnection().prepareStatement(sql);
        preparedStatement.setString(1,order.getOrderId());
        preparedStatement.setString(2,order.getItemName());
        preparedStatement.setInt(3,order.getCustomerId());
        preparedStatement.setBigDecimal(4,order.getPrice());
        preparedStatement.executeUpdate();
    }
}
